package com.my.kafka.serer;

import java.util.List;
import java.util.Optional;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;

/***
 * 消费者
 * @author my
 *
 */
@Component
public class DataConsumer {

	private Logger logger = LoggerFactory.getLogger(this.getClass());
	
	//private ConsumerConnector consumer;
	
	// 对于大多数的用户来说，这个是默认的，对于那些手动构建自定义容器的人来说，这是一个好的过渡路径
	@Autowired
	private ConcurrentKafkaListenerContainerFactory<String, String> consumerFactory;
	
	@Bean
	public KafkaListenerContainerFactory<?> batchFactory() {
		// 设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
		this.consumerFactory.setBatchListener(true);
		consumerFactory.setConcurrency(2);
		consumerFactory.getContainerProperties().setAckMode(AckMode.MANUAL);
		return this.consumerFactory;
	}
	
	@KafkaListener(topics = "#{'${kafka_topic}'.split(',')}", containerFactory = "batchFactory")
    public void messageListener(List<ConsumerRecord<String, String>> records, Acknowledgment ack){
		try {
			logger.info("MessageListener start...");
	        for (ConsumerRecord<?, String> record : records) {
	        	
	            Optional<?> kafkaKey = Optional.ofNullable(record.key());
	            if (kafkaKey.isPresent()) {
	            	logger.info("record key:" + record.key());
				} else {
					logger.info("no record key");
				}
	            
	            Optional<?> value = Optional.ofNullable(record.value());
	            logger.info("record value:" + value.get());
	        }

	        this.logger.info("MessageListener end...");
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			ack.acknowledge();
		}
    }
}
